package com.moli.iov.common.hbase.service.impl;

import com.moli.iov.common.hbase.dto.create.CreateTableCreateDto;
import com.moli.iov.common.hbase.dto.create.InsertBatchCreateDto;
import com.moli.iov.common.hbase.dto.create.InsertCreateDto;
import com.moli.iov.common.hbase.dto.modify.DeleteBatchRowModifyDto;
import com.moli.iov.common.hbase.dto.modify.DeleteRowModifyDto;
import com.moli.iov.common.hbase.dto.modify.DropTableModifyDto;
import com.moli.iov.common.hbase.dto.modify.TruncateTableModifyDto;
import com.moli.iov.common.hbase.dto.query.*;
import com.moli.iov.common.hbase.service.IHbaseService;
import com.moli.iov.util.SerializeUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;

/**
 * @ClassName HbaseServiceImpl
 * @Description TODO
 * @Author lixy
 * @Date 2019/4/22 11:55
 **/
@Slf4j
public class HBaseServiceImpl implements DisposableBean, IHbaseService {

    /**
     * HBase connection.
     */
    private Connection connection;

    public HBaseServiceImpl(Connection connection) {
        this.connection = connection;
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }

    /**
     * According to table name and scan and {@link RowBounds} filter of result list.
     * @param tableName Table name.
     * @param scan {@link Scan}.
     * @return After of Pagination filter result list.
     * @throws Exception {@link NullPointerException} or {@link Exception}.
     */
    private List<byte[]> selectList(String tableName, Scan scan) throws Exception {
        List<byte[]> results = new ArrayList<>();

        Table table = connection.getTable(TableName.valueOf(tableName));

        ResultScanner resultScanner = null;
        try {
            resultScanner = table.getScanner(scan);
            Iterator<Result> iterator = resultScanner.iterator();
            Result result = null;
            while (iterator.hasNext()) {
                result = iterator.next();
                byte[] value = result.getValue(FAMILY, QUALIFIER);
                results.add(value);
                if(log.isDebugEnabled()){
                    log.debug("Scan HBase rows of row key is {}", Bytes.toString(result.getRow()));
                }
            }
            return results;
        } catch (IOException e) {
            log.error("Scan HBase rows of found a failure cause: {}", e);
            return results;
        } finally {
            if (resultScanner != null) {
                resultScanner.close();
            }
        }
    }

    /**
     * According to table name and start row and end row filter of result map.
     * @param tableName Table name.
     * @return After of filter result map.
     * @throws Exception {@link NullPointerException} or {@link Exception}.
     */
    private Map<String, byte[]> mapRow(String tableName, Scan scan) throws Exception {
        Map<String, byte[]> resultMap = new LinkedHashMap<>();
        Table table = connection.getTable(TableName.valueOf(tableName));
        ResultScanner resultScanner = null;
        try {
            resultScanner = table.getScanner(scan);
            Iterator<Result> iterator = resultScanner.iterator();
            Result result = null;
            while (iterator.hasNext()) {
                result = iterator.next();
                byte [] rowKey = result.getRow(), value = result.getValue(FAMILY, QUALIFIER);
                resultMap.put(new String(rowKey), value);
                if(log.isDebugEnabled()){
                    log.debug("Scan HBase rows of row key is {}", Bytes.toString(result.getRow()));
                }
            }
            return resultMap;
        } catch (IOException e) {
            log.error("Scan HBase rows of found a failure cause: {}", e);
            return resultMap;
        } finally {
            if (resultScanner != null) {
                resultScanner.close();
            }
        }
    }

    /**
     * According to table name and start row and end row filter of result map.
     * @param tableName Table name.
     * @param pageSize Page size.
     * @return After of filter result map.
     * @throws Exception {@link NullPointerException} or {@link Exception}.
     */
    @Override
    public Map<String, byte[]> mapRow(String tableName, Integer pageSize) throws Exception {
        return mapRow(tableName, pageSize, false);
    }

    /**
     * According to table name and start row and end row filter of result map.
     * @param tableName Table name.
     * @param pageSize Page size.
     * @param reversed Whether reversed.
     * @return After of filter result map.
     * @throws Exception {@link NullPointerException} or {@link Exception}.
     */
    @Override
    public Map<String, byte[]> mapRow(String tableName, Integer pageSize, Boolean reversed) throws Exception{
        Scan scan = buildPaginationScan(pageSize, reversed);
        return mapRow(tableName, scan);
    }

    /**
     * According to table name and start row and end row filter of result map.
     * @param tableName Table name.
     * @param startRow Start row.
     * @param stopRow Stop row.
     * @return After of filter result map.
     * @throws Exception {@link NullPointerException} or {@link Exception}.
     */
    @Override
    public Map<String, byte[]> mapRow(String tableName, String startRow, String stopRow) throws Exception {
        return mapRow(tableName, startRow, stopRow, false);
    }

    /**
     * According to table name and start row and end row filter of result map.
     * @param tableName Table name.
     * @param startRow Start row.
     * @param stopRow Stop row.
     * @param reversed Whether reversed.
     * @return After of filter result map.
     * @throws Exception {@link NullPointerException} or {@link Exception}.
     */
    @Override
    public Map<String, byte[]> mapRow(String tableName, String startRow, String stopRow, Boolean reversed) throws Exception {
        Scan scan = buildScan(startRow, stopRow, reversed);
        return mapRow(tableName, scan);
    }

    /**
     * According to table name and row prefix and {@link RowBounds} filter of result list.
     * @param tableName Table name.
     * @param rowPrefix Row prefix.
     * @param rowBounds Pagination of row bounds.
     * @return After of Pagination filter result list.
     * @throws Exception {@link NullPointerException} or {@link Exception}.
     */
    @Override
    public List<byte[]> selectList(String tableName,
                                     byte[] rowPrefix, RowBounds rowBounds) throws Exception {

        if(rowPrefix == null){
            throw new NullPointerException("Scan HBase rows of row prefix not found.");
        }

        Scan scan = buildPaginationScan(rowBounds);
        scan.setRowPrefixFilter(rowPrefix);
        return selectList(tableName, scan);
    }

    /**
     * According to table name and start row and stop row and {@link RowBounds} filter of result list.
     * @param tableName Table name.
     * @param startRow Start row.
     * @param stopRow Stop row.
     * @param rowBounds Pagination of row bounds.
     * @return After of Pagination filter result list.
     * @throws Exception {@link NullPointerException} or {@link Exception}.
     */
    @Override
    public List<byte[]> selectList(String tableName,
                                    byte[] startRow, byte[] stopRow, RowBounds rowBounds) throws Exception {

        if(startRow == null || stopRow == null){
            throw new NullPointerException("Scan HBase rows of start row or stop row not found.");
        }

        Scan scan = buildPaginationScan(rowBounds);
        scan.setStartRow(startRow);
        scan.setStopRow(stopRow);
        return selectList(tableName, scan);
    }

    /**
     * According to row key and table name filter of single result.
     * @param tableName Table name.
     * @param rowKey Row key.
     * @return After of filter single result.
     * @throws Exception {@link NullPointerException} or {@link Exception}.
     */
    @Override
    public byte[] selectOne(String tableName, String rowKey) throws Exception {

        if(rowKey == null){
            throw new NullPointerException("Scan HBase rows of row key not found.");
        }

        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));

        Result result = null;
        try {
            result = table.get(get);
            return result.getValue(FAMILY, QUALIFIER);
        } catch (IOException e) {
            log.error("Scan HBase rows of found a failure cause: {}", e);
        }
        return null;
    }

    @Override
    public Boolean insertData(InsertCreateDto insertCreateDto) throws Exception {
        // log.info("insertData服务被调用-----------入参：{}",insertCreateDto.toString());
        Table table = connection.getTable(TableName.valueOf(insertCreateDto.getTableName()));
        Put put = new Put(insertCreateDto.getRowKey().getBytes());
        Set<HashMap.Entry<String, Object>> set = insertCreateDto.getColumnAndValue().entrySet();
        Iterator<HashMap.Entry<String, Object>> iterator = set.iterator();
        while (iterator.hasNext()) {
            HashMap.Entry<String, Object> next = iterator.next();
            String qualifier = next.getKey();
            Object value = next.getValue();
            byte[] realValue = null;
            if (!(value instanceof byte[])) {
                realValue = SerializeUtils.fstSerialize(value);
            } else {
                realValue = (byte[]) value;
            }
            put.add(insertCreateDto.getFamily().getBytes(), qualifier.getBytes(), realValue);

        }
        try {
            table.put(put);
            table.close();
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("tableName:" + insertCreateDto.getTableName() + "rowkey:" + insertCreateDto.getRowKey() + "family:" + insertCreateDto.getFamily() + "   insert fail...");
            return false;
        }
    }

    @Override
    public Boolean insertBatchData(InsertBatchCreateDto insertBatchCreateDto) throws Exception {
        //log.info("insertBatchData服务被调用-----------入参：{}",insertBatchCreateDto.toString());
        Table table = connection.getTable(TableName.valueOf(insertBatchCreateDto.getTableName()));
        List<Put> puts = new ArrayList<>(0);
        Iterator<HashMap.Entry<String, HashMap<String, Object>>> iterator = insertBatchCreateDto.getRowKeyAndColumnAndValue().entrySet().iterator();
        while (iterator.hasNext()) {
            HashMap.Entry<String, HashMap<String, Object>> next = iterator.next();
            //行键
            String rowKey = next.getKey();
            //内容
            HashMap<String, Object> valueMap = next.getValue();
            Put put = new Put(rowKey.getBytes());
            Iterator iteratorValue = valueMap.entrySet().iterator();
            while (iteratorValue.hasNext()) {
                Map.Entry entry = (Map.Entry) iteratorValue.next();
                String qualifier = String.valueOf(entry.getKey());
                Object value = entry.getValue();
                byte[] realValue = null;
                if (!(value instanceof byte[])) {
                    realValue = SerializeUtils.fstSerialize(value);
                } else {
                    realValue = (byte[]) value;
                }
                put.add(insertBatchCreateDto.getFamily().getBytes(), qualifier.getBytes(), realValue);
                puts.add(put);
            }
        }

        try {
            //插入信息
            table.put(puts);
            table.close();
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            log.error(" insert data error......");
            return false;
        }
    }

    @Override
    public Boolean createTable(CreateTableCreateDto createTableCreateDto) throws Exception {
        log.info("createTable服务被调用-----------入参：{}", createTableCreateDto.toString());
        try {

            Admin hBaseAdmin = connection.getAdmin();
            if (hBaseAdmin.tableExists(TableName.valueOf(createTableCreateDto.getTableName()))) {
                // 如果存在要创建的表，那么先删除，再创建
                hBaseAdmin.disableTable(TableName.valueOf(createTableCreateDto.getTableName()));
                hBaseAdmin.deleteTable(TableName.valueOf(createTableCreateDto.getTableName()));
            }
            HTableDescriptor tableDescriptor = new HTableDescriptor(createTableCreateDto.getTableName());
            tableDescriptor.addFamily(new HColumnDescriptor(createTableCreateDto.getFamily()));
            hBaseAdmin.createTable(tableDescriptor);
            hBaseAdmin.close();
            return true;
        } catch (MasterNotRunningException e) {
            e.printStackTrace();
            log.error("tableName:" + createTableCreateDto.getTableName() + " family:" + createTableCreateDto.getFamily() + " create fail....");
            return false;
        } catch (ZooKeeperConnectionException e) {
            e.printStackTrace();
            log.error("tableName:" + createTableCreateDto.getTableName() + " family:" + createTableCreateDto.getFamily() + " create fail....");
            return false;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("tableName:" + createTableCreateDto.getTableName() + " family:" + createTableCreateDto.getFamily() + " create fail....");
            return false;
        }
    }

    @Override
    public Boolean truncateTable(TruncateTableModifyDto truncateTableModifyDto) throws Exception {

        log.info("truncateTable服务被调用-----------入参：{}", truncateTableModifyDto.toString());
        try {
            Admin hBaseAdmin = connection.getAdmin();
            TableName table = TableName.valueOf(truncateTableModifyDto.getTableName());
            hBaseAdmin.disableTable(table);
            hBaseAdmin.truncateTable(table, true);
            hBaseAdmin.close();
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public Boolean dropTable(DropTableModifyDto dropTableModifyDto) throws Exception {
        log.info("dropTable服务被调用-----------入参：{}", dropTableModifyDto.toString());
        String tableName = dropTableModifyDto.getTableName();
        try {
            Admin admin = connection.getAdmin();
            admin.disableTable(TableName.valueOf(dropTableModifyDto.getTableName()));
            admin.deleteTable(TableName.valueOf(dropTableModifyDto.getTableName()));
            admin.close();
            return true;
        } catch (MasterNotRunningException e) {
            e.printStackTrace();
            log.error("tableName:" + tableName + " drop fail...");
            return false;
        } catch (ZooKeeperConnectionException e) {
            e.printStackTrace();
            log.error("tableName:" + tableName + " drop fail...");
            return false;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("tableName:" + tableName + " drop fail...");
            return false;
        }
    }

    @Override
    public Boolean deleteRow(DeleteRowModifyDto deleteRowModifyDto) throws Exception {
        log.info("deleteRow服务被调用-----------入参：{}", deleteRowModifyDto.toString());
        try {
            Table table = connection.getTable(TableName.valueOf(deleteRowModifyDto.getTableName()));
            List list = new ArrayList();
            Delete d1 = new Delete(deleteRowModifyDto.getRowKey().getBytes());
            list.add(d1);
            table.delete(list);
            log.info("删除行成功!");
            table.close();
            return true;

        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public Boolean deleteBatchRow(DeleteBatchRowModifyDto deleteBatchRowModifyDto) throws Exception {

        log.info("deleteBatchRow服务被调用-----------入参：{}", deleteBatchRowModifyDto.toString());
        try {
            ArrayList list = new ArrayList();
            Table table = connection.getTable(TableName.valueOf(deleteBatchRowModifyDto.getTableName()));
            for (String rowkey : deleteBatchRowModifyDto.getRowKeys()) {
                Delete d1 = new Delete(rowkey.getBytes());
                list.add(d1);
            }
            table.delete(list);
            log.info("删除行成功!");
            table.close();
            return true;

        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public Boolean queryRow(QueryRowQueDto queryRowQueDto) throws Exception {
        //log.info("queryRow服务被调用-----------入参：{}",queryRowQueDto.toString());

        try {
            Table table = connection.getTable(TableName.valueOf(queryRowQueDto.getTableName()));
            Get get = new Get(queryRowQueDto.getRowKey().getBytes());
            Result result = table.get(get);
            if (null != result) {
                byte[] values = result.getValue("fm1".getBytes(), "messageContent".getBytes());
                List<Cell> listCells = result.listCells();
                for (Cell cell : listCells) {
                    byte[] familyArray = cell.getFamilyArray();
                    byte[] qualifierArray = cell.getQualifierArray();
                    byte[] valueArray = cell.getValueArray();
                }

            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    @Override
    public HashMap<String, byte[]> getResultScanner(GetResultScannerQueDto getResultScannerQueDto) throws Exception {
        long startTimes = System.currentTimeMillis();
        log.info("起始时间-----------------》{}", startTimes);
        log.info("getResultScanner查询服务被调用-----------入参：{}", getResultScannerQueDto.toString());
        HashMap<String, byte[]> hashMap = new HashMap<>(0);
        ResultScanner rs = null;
        Scan scan = new Scan();
        if (!StringUtils.isEmpty(getResultScannerQueDto.getStartRow())) {
            scan.setStartRow(getResultScannerQueDto.getStartRow().getBytes());
        }
        if (!StringUtils.isEmpty(getResultScannerQueDto.getStopRow())) {
            scan.setStopRow(getResultScannerQueDto.getStopRow().getBytes());
        }

        if (!CollectionUtils.isEmpty(getResultScannerQueDto.getFilters())) {
            for (Filter filter : getResultScannerQueDto.getFilters()) {
                scan.setFilter(filter);
            }
        }
        Table table = connection.getTable(TableName.valueOf(getResultScannerQueDto.getTableName()));
        scan.addFamily(FAMILY);
        scan.setCacheBlocks(true);
        scan.setCaching(8000);
        scan.setBatch(2);
        try {
            rs = table.getScanner(scan);
            for (Result result : rs) {
                String rowStr = new String(result.getRow());
                byte[] value = result.getValue(getResultScannerQueDto.getFamily()==null?"fm1".getBytes():getResultScannerQueDto.getFamily().getBytes(), getResultScannerQueDto.getColumn().getBytes());
                hashMap.put(rowStr, value);
            }
            long endTimes = System.currentTimeMillis();
            log.info("结束时间---------》{}", endTimes);
            log.info("查询共用时-----------》{}", endTimes - startTimes);
            log.info("getResultScanner查询服务结果-----------返回值数量：{}", hashMap.size());

            return hashMap;
        } catch (IOException e) {
            e.printStackTrace();
            log.error(getResultScannerQueDto.getTableName() + "表扫描异常....");
        } finally {
            if (rs != null) {
                rs.close();
            }
        }
        return hashMap;

    }

    Scan assemblyConditions(GetResultScanner scan){
        if(scan==null){
            return null;
        }
        Scan scanResult = new Scan();

        if(!StringUtils.isEmpty(scan.getVin())){
            if (scan.getEndTime()!=null && !StringUtils.isEmpty(scan.getVin())) {
                scanResult.setStartRow(getRowKey(scan.getVin(),scan.getEndTime()));
            }else{
                scanResult.setRowPrefixFilter(Bytes.toBytes(scan.getVin()+"_"));
            }
            if (scan.getStartTime()!=null && !StringUtils.isEmpty(scan.getVin())) {
                scanResult.setStopRow(getRowKey(scan.getVin(),scan.getStartTime()));
            }
        }

        if(scan.getPageSize()!=null){
            Filter filter = new PageFilter(scan.getPageSize());
            scanResult.setFilter(filter);
        }
        return scanResult;
    }

    @Override
    public List<byte[]> getDataResult(GetResultScanner resultScanner) throws Exception {
        List<byte[]> list = new ArrayList<>();

        Scan scan = assemblyConditions(resultScanner);
        Table table = connection.getTable(TableName.valueOf(resultScanner.getTableName()));
//        scan.addFamily("fm1".getBytes());
        scan.addColumn(FAMILY, QUALIFIER);
//        scan.setBatch(3);

        ResultScanner rs = null;
        try {

            rs = table.getScanner(scan);
            Iterator<Result> it = rs.iterator();
//            rs.close();
            while (it.hasNext()) {
                Result result = it.next();
                byte[] value = result.getValue(FAMILY, QUALIFIER);
                System.out.println("Rowkey:"+Bytes.toString(result.getRow()));
                list.add(value);
            }

            return list;
        } catch (IOException e) {
            e.printStackTrace();
            return list;
        } finally {
            if (rs != null) {
                rs.close();
            }
        }
    }

    @Override
    public List<byte[]> getDataResult(String startRow, String endRow, String tableName) throws Exception {
        List<byte[]> list = new ArrayList<>();

        Scan scan = new Scan();
        if (!StringUtils.isEmpty(startRow)) {
            scan.setStartRow(Bytes.toBytes(startRow));
        }
        if (!StringUtils.isEmpty(endRow)) {
            scan.setStopRow(Bytes.toBytes(endRow));
        }

        Table table = connection.getTable(TableName.valueOf(tableName));
//        scan.addFamily("fm1".getBytes());
        scan.addColumn(FAMILY, QUALIFIER);
        scan.setCacheBlocks(true);
        scan.setCaching(100000);
//        scan.setBatch(3);

        ResultScanner rs = null;
        try {

            rs = table.getScanner(scan);
            Iterator<Result> it = rs.iterator();
//            rs.close();
            while (it.hasNext()) {
                Result result = it.next();
                byte[] value = result.getValue(FAMILY, QUALIFIER);
                list.add(value);
            }

            return list;
        } catch (IOException e) {
            e.printStackTrace();
            return list;
        } finally {
            if (rs != null) {
                rs.close();
            }
        }
    }


    @Override
    public Integer count(CountQueDto countQueDto) throws Exception {
        log.info("count服务被调用-----------入参：{}", countQueDto.toString());
        Table table = connection.getTable(TableName.valueOf(countQueDto.getTableName()));
        Scan scan = new Scan();
        scan.setCaching(5000);
        scan.setCacheBlocks(true);
        scan.addFamily(FAMILY);
        Filter filter = new FirstKeyOnlyFilter();
        scan.setFilter(filter);
        int initnum = 0;
        try {
            long begintime = System.currentTimeMillis();
            log.info("------------------countbegintime:" + begintime);
            ResultScanner results = table.getScanner(scan);
            long endtime = System.currentTimeMillis();
            log.info("------------------countendtime:" + endtime);
            log.info("------------------counttime:" + (endtime - begintime));
            for (Result result : results) {
                ++initnum;
            }
            log.info("-------------->initnum={}", initnum);
            return initnum;
        } catch (IOException e) {
            e.printStackTrace();
            return initnum;
        }
    }

    @Override
    public HashMap<String, byte[]> queryResult(QueryResultQueDto queryResultQueDto) throws Exception {
        log.info("queryResult服务被调用-----------入参：{}", queryResultQueDto.toString());
        HashMap<String, byte[]> map = new HashMap<String, byte[]>(0);
        Table table = connection.getTable(TableName.valueOf(queryResultQueDto.getTableName()));
        Scan scan = new Scan();
        scan.setCaching(5000);
        scan.setCacheBlocks(true);
        String regx = ".*" + queryResultQueDto.getRowLike() + ".*";
        Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(regx));
        scan.setFilter(filter);
        try {
            ResultScanner resultScanner = table.getScanner(scan);
            for (Result result : resultScanner) {
                String rowStr = new String(result.getRow());
                byte[] value = result.getValue(queryResultQueDto.getFamily().getBytes(), queryResultQueDto.getColumn().getBytes());
                map.put(rowStr, value);
            }
            return map;
        } catch (IOException e) {
            e.printStackTrace();
            return map;
        }

    }

    @Override
    public void destroy() throws Exception {
        if(!connection.isClosed()){
            connection.close();
        }
    }
}
